use crate::config::Config;
use crate::errors::{BinanceError, Result};
use crate::ws::ws_client::WsClient;

use serde_json::from_value;
use serde_json::Value;
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};

pub use crate::ws::{
    AggTrade, AggTradeRaw, DepthUpdate, DepthUpdateRaw, ForceOrderEvent, ForceOrderRaw, Kline,
    KlineRaw, MarketStreamEvent,
};
use crate::ws::market_types::{
    BookTickerEvent, BookTickerRaw, MarkPriceEvent, MarkPriceRaw,
};

async fn create_ws_channel(
    url: String,
    proxy: Option<String>,
) -> Result<(WsClient, UnboundedReceiver<Value>)> {
    let (client, rx) = WsClient::new(&url, proxy).await
        .map_err(|e| BinanceError::Other(format!("{:?}", e)))?;
    Ok((client, rx))
}

macro_rules! subscribe_single_stream {
    ($func_name:ident, $event_type:ty, $raw_type:ty, $topic_builder:expr) => {
        pub async fn $func_name(
            symbol: &str,
            config: &Config,
        ) -> Result<UnboundedReceiver<$event_type>> {
            let topic = $topic_builder(symbol);
            let url = format!("{}/ws/{}", config.ws_base_url(), topic);
            let (client, mut raw_rx) = create_ws_channel(url, config.proxy.clone()).await?;
            let (tx, rx) = unbounded_channel();

            tokio::spawn(async move {
                while let Some(raw) = raw_rx.recv().await {
                    if let Ok(raw_data) = serde_json::from_value::<$raw_type>(raw) {
                        if let Some(parsed) = raw_data.to_model() {
                            let _ = tx.send(parsed);
                        }
                    }
                }

                drop(client); // 防止 WsClient 被泄漏
            });

            Ok(rx)
        }
    };
}

subscribe_single_stream!(
    subscribe_book_ticker,
    BookTickerEvent,
    BookTickerRaw,
    |symbol: &str| format!("{}@bookTicker", symbol.to_lowercase())
);

subscribe_single_stream!(
    subscribe_mark_price,
    MarkPriceEvent,
    MarkPriceRaw,
    |symbol: &str| format!("{}@markPrice@1s", symbol.to_lowercase())
);

subscribe_single_stream!(
    subscribe_force_order,
    ForceOrderEvent,
    ForceOrderRaw,
    |symbol: &str| format!("{}@forceOrder", symbol.to_lowercase())
);

subscribe_single_stream!(
    subscribe_agg_trade,
    AggTrade,
    AggTradeRaw,
    |symbol: &str| format!("{}@aggTrade", symbol.to_lowercase())
);

pub async fn subscribe_kline(
    symbol: &str,
    interval: &str,
    config: &Config,
) -> Result<UnboundedReceiver<Kline>> {
    let topic = format!("{}@kline_{}", symbol.to_lowercase(), interval);
    let url = format!("{}/ws/{}", config.ws_base_url(), topic);

    let (client, mut raw_rx) = WsClient::new(&url, config.proxy.clone()).await
        .map_err(|e| BinanceError::Other(e.to_string()))?;
    let (tx, rx) = unbounded_channel();

    tokio::spawn(async move {
        while let Some(raw) = raw_rx.recv().await {
            if let Ok(raw_kline) = serde_json::from_value::<KlineRaw>(raw) {
                if let Some(kline) = raw_kline.to_model() {
                    let _ = tx.send(kline);
                }
            }
        }
        drop(client);
    });

    Ok(rx)
}

/// 订阅 @depth@100ms，输出标准化订单簿更新（DepthUpdate）
pub async fn subscribe_depth_update(
    symbol: &str,
    config: &Config,
    speed: Option<&str>,
) -> Result<UnboundedReceiver<DepthUpdate>> {
    let suffix = speed.unwrap_or("250ms");
    let topic = format!("{}@depth@{}", symbol.to_lowercase(), suffix);
    let url = format!("{}/ws/{}", config.ws_base_url(), topic);
    let (client, mut raw_rx) = create_ws_channel(url, config.proxy.clone()).await?;
    let (tx, rx) = unbounded_channel();

    tokio::spawn(async move {
        while let Some(raw) = raw_rx.recv().await {
            if let Ok(raw_data) = serde_json::from_value::<DepthUpdateRaw>(raw) {
                if let Some(depth) = raw_data.to_model() {
                    let _ = tx.send(depth);
                }
            }
        }

        drop(client);
    });

    Ok(rx)
}

/// 订阅 !bookTicker（全市场）
pub async fn subscribe_all_book_ticker(
    config: &Config,
) -> Result<UnboundedReceiver<BookTickerEvent>> {
    let url = format!("{}/ws/!bookTicker", config.ws_base_url());
    let (client, mut raw_rx) = create_ws_channel(url, config.proxy.clone()).await?;
    let (tx, rx) = unbounded_channel();

    tokio::spawn(async move {
        while let Some(raw) = raw_rx.recv().await {
            if let Ok(raw) = serde_json::from_value::<BookTickerRaw>(raw) {
                if let Some(quote) = raw.to_model() {
                    let _ = tx.send(quote);
                }
            }
        }

        drop(client);
    });

    Ok(rx)
}

/// 多路复用订阅接口（用于 batch）
pub async fn subscribe_multi_streams(
    streams: &[String],
    config: &Config,
) -> Result<UnboundedReceiver<MarketStreamEvent>> {
    let stream_param = streams.join("/");
    let url = format!("{}/stream?streams={}", config.ws_base_url(), stream_param);

    let (client, mut raw_rx) = create_ws_channel(url, config.proxy.clone()).await?;
    let (tx, rx) = unbounded_channel();

    tokio::spawn(async move {
        while let Some(raw_msg) = raw_rx.recv().await {
            if let Ok(v) = serde_json::from_value::<Value>(raw_msg) {
                if let (Some(data), Some(event_type)) = (
                    v.get("data").cloned(),
                    v.get("data").and_then(|d| d.get("e")).and_then(|e| e.as_str()),
                ) {
                    let parsed = match event_type {
                        "aggTrade" => serde_json::from_value::<AggTradeRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::AggTrade),
                        "depthUpdate" => serde_json::from_value::<DepthUpdateRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::DepthUpdate),
                        "kline" => serde_json::from_value::<KlineRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::Kline),
                        "bookTicker" => serde_json::from_value::<BookTickerRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::BookTicker),
                        "markPriceUpdate" | "markPrice" => serde_json::from_value::<MarkPriceRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::MarkPrice),
                        "forceOrder" => serde_json::from_value::<ForceOrderRaw>(data)
                            .ok()
                            .and_then(|r| r.to_model())
                            .map(MarketStreamEvent::ForceOrder),
                        _ => None,
                    };

                    if let Some(event) = parsed {
                        let _ = tx.send(event);
                    }
                }
            }
        }

        drop(client);
    });

    Ok(rx)
}
/// 拆分 stream 列表，按最大 URL 长度限制分批（防止请求被拒）
fn split_streams_by_max_len(streams: &[String], max_len: usize) -> Vec<Vec<String>> {
    let mut batches = Vec::new();
    let mut current_batch = Vec::new();
    let mut current_length = 0;

    for stream in streams {
        let added_len = if current_batch.is_empty() {
            stream.len()
        } else {
            stream.len() + 1 // 分隔符 "/"
        };

        if current_length + added_len > max_len {
            if !current_batch.is_empty() {
                batches.push(current_batch);
            }
            current_batch = vec![stream.clone()];
            current_length = stream.len();
        } else {
            current_batch.push(stream.clone());
            current_length += added_len;
        }
    }

    if !current_batch.is_empty() {
        batches.push(current_batch);
    }

    // 打印统计信息
    let total_streams: usize = batches.iter().map(|b| b.len()).sum();
    println!(
        "split_streams_by_max_len: total {} streams, split into {} batches, max_len={}",
        total_streams,
        batches.len(),
        max_len
    );
    for (i, batch) in batches.iter().enumerate() {
        let length = batch.iter().map(|s| s.len()).sum::<usize>() + batch.len().saturating_sub(1);
        println!("Batch {}: {} streams, total length {}", i + 1, batch.len(), length);
    }

    batches
}

/// 批量订阅多个流（自动分批避免超长 URL），统一输出 MarketStreamEvent 枚举
pub async fn batch_subscribe_multi_streams(
    streams: &[String],
    config: &Config,
    max_url_len: usize,
) -> Result<UnboundedReceiver<MarketStreamEvent>> {
    let (tx, rx) = unbounded_channel();
    let batches = split_streams_by_max_len(streams, max_url_len);

    for batch in batches {
        let stream_param = batch.join("/");
        let url = format!("{}/stream?streams={}", config.ws_base_url(), stream_param);

        let (client, mut raw_rx) = WsClient::new(&url, config.proxy.clone())
            .await
            .map_err(|e| BinanceError::Other(format!("WebSocket connect failed: {}", e)))?;

        let internal_tx = tx.clone();
        tokio::spawn(async move {
            while let Some(raw_msg) = raw_rx.recv().await {
                let Ok(json) = from_value::<Value>(raw_msg) else {
                    continue;
                };

                let data = json.get("data").cloned();
                let event_type = data
                    .as_ref()
                    .and_then(|d| d.get("e"))
                    .and_then(|e| e.as_str());

                let parsed = match (event_type, data.clone()) {
                    (Some("bookTicker"), Some(d)) => {
                        from_value::<BookTickerRaw>(d).unwrap().to_model().map(MarketStreamEvent::BookTicker)
                    }
                    (Some("aggTrade"), Some(d)) => {
                        from_value::<AggTradeRaw>(d).unwrap().to_model().map(MarketStreamEvent::AggTrade)
                    }
                    (Some("depthUpdate"), Some(d)) => {
                        from_value::<DepthUpdateRaw>(d).unwrap().to_model().map(MarketStreamEvent::DepthUpdate)
                    }
                    (Some("kline"), Some(d)) => {
                        from_value::<KlineRaw>(d).ok().unwrap().to_model().map(MarketStreamEvent::Kline)
                    }
                    (Some("markPriceUpdate" | "markPrice"), Some(d)) => {
                        from_value::<MarkPriceRaw>(d).ok().unwrap().to_model().map(MarketStreamEvent::MarkPrice)
                    }
                    (Some("forceOrder"), Some(d)) => {
                        from_value::<ForceOrderRaw>(d).ok().unwrap().to_model().map(MarketStreamEvent::ForceOrder)
                    }
                    _ => None,
                };

                if let Some(event) = parsed {
                    let _ = internal_tx.send(event);
                }
            }

            drop(client); // 清理连接引用
        });
    }

    Ok(rx)
}